package cgl.granules.application;

import java.util.Enumeration;
import java.io.IOException;
import java.util.Properties;
import java.io.FileWriter;

import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceBase;
import cgl.granules.operation.MapReduceException;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;
import cgl.granules.results.ResultsFactory;
import cgl.narada.event.NBEvent;

/**
 * An example demonstrating the reduce role within a dummy pipeline operation.
 * 
 * @author Chao Sun
 * 
 */
public class DummyReduce extends MapReduceBase {
	private String reducenum;
	private int filenum = 0;
	String workerType = "Unknown ";
	String filenamebase = "/globalhome/chaosun/test/outputfiles/file_";

	public DummyReduce() {
		setDomain("Generic");
		setVersionInformation(200);
		setOperationIdentifier("DummyPipeline-ReduceFunction-GranulesSample");
	}

	public void execute() {
		try {
			StreamingAccess mapReduceDataset = getMapReduceDataset();

			Properties props = getProcessingDirectives();
			String property = "Reducer Number";
			if (props.containsKey(property)) {
				workerType = property + ": " + props.getProperty(property);
			} else {
				System.out.println("Reducer info incomplete! existing ...");
				return;
			}
			reducenum = props.getProperty(property);

			while (mapReduceDataset.isDataAvailable()) {
				NBEvent nbEvent = mapReduceDataset.getStreamingData();

				processResultsFromMap(nbEvent);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private synchronized void processResultsFromMap(NBEvent nbEvent)
			throws IOException, MarshallingException, MapReduceException {
		String synopsis = (String) nbEvent.getContentSynopsis();

		byte[] resultBytes = nbEvent.getContentPayload();
		Results results = ResultsFactory.getInstance().getResults(resultBytes);
		byte[] reportBytes = results.getResultPayload();

		ResultPayload resultPayload = ResultPayload.getInstance();

		//String printst = new String(reportBytes);
		//System.out.println("paload bytes: " + printst);

		try {
			//System.out.println("reducer [" + reducenum + "] converting data");
			String filebytes[] = resultPayload.getStringArray(reportBytes);

			int fileN = filebytes.length;
			//System.out.println("reducer [" + reducenum + "] data containing "
			//+ fileN + " files");

			for (int i = 0; i < fileN; i++) {
				String filename = filenamebase + reducenum + "_" + filenum
						+ ".txt";
				FileWriter filewriter = new FileWriter(filename);
				filewriter.write(filebytes[i]);
				filewriter.flush();
				filewriter.close();
				System.out.println("reducer [" + reducenum + "] saved file ["
						+ filenum + "]");
				filenum++;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

		setReceivedOutputFrom(results.getInstanceIdentifier());

		if (!isAwaitingOutputs()) {
			setTeminationConditionReached();
			System.out.println("\n" + synopsis + ":-> "
					+ "mission finished by reducer [" + reducenum + "]");
		}
	}
}